package com.dayu.tcp.service.client;

import com.dayu.tcp.exception.SocketRuntimeException;
import com.dayu.tcp.listener.BeforeMessageEventListener;
import com.dayu.tcp.logging.Log;
import com.dayu.tcp.logging.LogFactory;
import com.dayu.tcp.protocol.K1RestfulRequest;
import com.dayu.tcp.protocol.K1RestfulResponse;
import com.dayu.tcp.protocol.codec.K1Decoder;
import com.dayu.tcp.protocol.codec.K1Encoder;
import com.dayu.tcp.service.EventDispatcher;
import com.dayu.tcp.service.Service;
import com.dayu.tcp.service.SocketType;
import com.dayu.tcp.service.WrappedChannel;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Getter;
import lombok.Setter;

import java.net.SocketAddress;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * @author yehangtech
 * @date 2025/4/13 14:40
 * @description 统一的客户端封装
 */
public class BaseClient extends Service {

    private static final Log log = LogFactory.getLog(BaseClient.class);

    /**
     * 建立连接的超时时间（毫秒）,默认5000
     */
    @Getter
    @Setter
    protected long connectTimeout = 5000L;

    /**
     * 同步调用默认超时时间（毫秒）,默认5000
     */
    @Getter
    @Setter
    protected long syncInvokeTimeout = 5000L;

    protected ClientDispatchHandler dispatchHandler;

    @Getter
    @Setter
    protected SocketAddress curServer;

    @Getter
    @Setter
    protected WrappedChannel channel;

    /**
     * 用于连接的时候同步等待
     */
    protected Semaphore semaphore = new Semaphore(0);

    protected EventLoopGroup group;

    protected Bootstrap bootstrap;

    public BaseClient() {
        super();
    }

    @Override
    protected void init() {
        super.init();
        eventDispatcher = new EventDispatcher(this);
        dispatchHandler = new ClientDispatchHandler(eventDispatcher);
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        this.addEventListener(new BeforeMessageEventListener());
    }

    @Override
    public void shutdown() {
        if (null != group) {
            group.shutdownGracefully();
        }
    }

    public ChannelFuture connect(final SocketAddress socketAddress) {
        return this.connect(socketAddress, true);
    }

    public ChannelFuture connect(final SocketAddress socketAddress, boolean sync) {
        this.init();
        bootstrap.option(ChannelOption.TCP_NODELAY, tcpNoDelay);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, keepAlive);
        bootstrap.group(group).channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                //1: 只有TCP模式下才选择去注册编解码器
                if (Objects.equals(socketType, SocketType.NORMAL)) {
                    pipeline.addLast("K1Decoder", new K1Decoder());
                    pipeline.addLast("K1Encoder", K1Encoder.getSingleton());
                }
                //2: 注册各种自定义Handler
                LinkedHashMap<String, ChannelHandler> channelHandlerMap = getChannelHandlerMap();
                for (Map.Entry<String, ChannelHandler> entry : channelHandlerMap.entrySet()) {
                    pipeline.addLast(entry.getKey(), entry.getValue());
                }
                //3: 事件分发器
                pipeline.addLast("ClientDispatchHandler", dispatchHandler);
            }
        });
        return doConnect(socketAddress, sync);
    }

    /**
     * 客户端连接server
     *
     * @param socketAddress server的地址
     * @param sync          是否同步，true-是
     * @return
     */
    protected ChannelFuture doConnect(final SocketAddress socketAddress, boolean sync) {
        this.curServer = socketAddress;
        try {
            ChannelFuture future = bootstrap.connect(curServer).sync();
            this.addFutureEventListener((ChannelFutureListener) ch -> {
                if (ch.isSuccess()) {
                    this.channel = new WrappedChannel(ch.channel());
                    if (log.isDebugEnabled()) {
                        log.debug("Connected to '{" + socketAddress + "}' success.");
                    }
                } else {
                    log.error("Connect to '{" + socketAddress + "}' failed. cause by: ", ch.cause());
                }
                semaphore.release(Integer.MAX_VALUE - semaphore.availablePermits());
            });
            futureEventListenerList.forEach(future::addListener);
            future.channel().closeFuture();
            if (sync) {
                Throwable cause = null;
                try {
                    if (!semaphore.tryAcquire(connectTimeout, TimeUnit.MILLISECONDS)) {
                        cause = new SocketRuntimeException("time out.");
                    }
                } catch (InterruptedException e) {
                    throw new SocketRuntimeException(e);
                }
                if (cause != null) {
                    throw new SocketRuntimeException(cause);
                }
            }
            return future;
        } catch (Exception e) {
            throw new SocketRuntimeException(e);
        }
    }

    /**
     * 异步发送消息
     *
     * @param msg
     * @return
     */
    public ChannelFuture send(Object msg) {
        if (channel == null) {
            throw new SocketRuntimeException("channel have not connected.");
        }
        return channel.send(msg);
    }

    /**
     * 同步发送消息
     * @param msg
     * @return
     */
    public K1RestfulResponse sendSync(K1RestfulRequest msg) {
        return this.sendSync(msg, syncInvokeTimeout);
    }

    public K1RestfulResponse sendSync(K1RestfulRequest msg, long timeout) {
        if (channel == null) {
            throw new SocketRuntimeException("channel have not connected.");
        }
        return channel.sendSync(msg, timeout, TimeUnit.MILLISECONDS);
    }
}
